package sql;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.sql.Timestamp;

/**
 grouping sets用法
 在一个窗口内同时计算不同维度的聚合结果


 */
public class D9_4KafakSource_Tumble_Groupingsets {


    public static void main(String[] args) {


        Configuration flinkConf = new Configuration();
        flinkConf.setString("rest.port","9091");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(flinkConf);
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        String genSql = "CREATE TABLE ods_tb ( " +
            " stime STRING," +
            " name STRING," +
            " val BIGINT," +
            " rowtime AS to_timestamp(stime)," +
            " WATERMARK FOR rowtime AS rowtime - interval '10' second" +
            ") WITH ( " +
            "  'connector' = 'kafka'," +
            "  'topic' = 'test'," +
            "  'properties.bootstrap.servers' = 'wsl:9092'," +
            "  'properties.group.id' = 'testGroup'," +
            "  'scan.startup.mode' = 'latest-offset'," +
            "  'format' = 'json'" +
            ")";

        String print = "CREATE TABLE print (" +"    " +
            "    window_start timestamp, " +
            "    window_end timestamp, " +
            "    name STRING, " +
            "    val BIGINT " +
            ") WITH (" +
            "     'connector' = 'print'" +
            ")";

        String sql = "insert into print " +
                " SELECT " +
            " window_start, window_end," +
            " name," +
            " sum(val) " +
            "FROM " +
            " table(tumble(table ods_tb, descriptor(rowtime), interval '1' minutes))" +
            " GROUP BY window_start,window_end,window_time" +
            " ," +
            " GROUPING SETS(" +
            " ()," +        //  计算总和
            " (name)" +     // 按name聚合计算总和
            ")";
        System.out.println(sql);
        /**
         *
         * output: 没有聚合字段的位置置为null,
         *
         * +I[2024-08-02T12:00, 2024-08-02T12:01, bb, 6]
         * +I[2024-08-02T12:00, 2024-08-02T12:01, null, 11]
         * +I[2024-08-02T12:00, 2024-08-02T12:01, aa, 5]
         * +I[2024-08-02T12:01, 2024-08-02T12:02, aa, 6]
         * +I[2024-08-02T12:01, 2024-08-02T12:02, bb, 6]
         * +I[2024-08-02T12:01, 2024-08-02T12:02, null, 12]
         *
         *
         * so 聚合字段里本身有null的话就会产生歧义
         * +I[2024-08-02T12:02, 2024-08-02T12:03, aa, 6]
         * +I[2024-08-02T12:02, 2024-08-02T12:03, bb, 6]
         * +I[2024-08-02T12:02, 2024-08-02T12:03, null, 6]
         * +I[2024-08-02T12:02, 2024-08-02T12:03, null, 18]
         */


        tableEnv.executeSql(genSql);
        tableEnv.executeSql(print);

        System.out.println(sql);
        tableEnv.executeSql(sql);


        System.out.println(new Timestamp(System.currentTimeMillis()));



    }
}
